package com.flute.haflute.agent.manage;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.List;

import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.agent.job.JobResourceManager;
import com.flute.haflute.agent.topo.ClusterTopoDBService;
import com.flute.haflute.agent.topo.Node;
import com.flute.haflute.common.ClusterConstants;
import com.flute.haflute.common.ClusterContext;
import com.flute.haflute.net.messaging.Message;
import com.flute.haflute.net.messaging.MessageFactory;
import com.flute.haflute.net.messaging.MessagingService;

public class ClusterResourceCenter {

	private static Logger logger = LoggerFactory
			.getLogger(ClusterResourceCenter.class);

	public static synchronized void removeNode(Node node) {

		ClusterTopoDBService.getInstance().removeNode(node);

		JobResourceManager.getInstance().releaseNodes4Job(node);

	}

	public static boolean addNode(Node node) {

		return true;
	}

	public static synchronized void endJobManually(Node jobController) {

		List<Node> shoudNotifyNodes = JobResourceManager.getInstance()
				.releaseNodes4Job(jobController);

		try {
			Message message = MessageFactory.getShutdownChildMessage(
					ClusterConstants.JOB_MASTER_TAG, new byte[0]);

			// send message to Master to shutdown
			IoSession session = MessagingService.getInstance().send2Node(
					new InetSocketAddress(jobController.getIp(),
							ClusterContext.getHeartBeatPort()), message,
					new IoHandlerAdapter());

			session.close(false).awaitUninterruptibly();

			// send message to Slaves to shutdown
			message = MessageFactory.getShutdownChildMessage(
					ClusterConstants.JOB_SLAVE_TAG, new byte[0]);

			for (Node node : shoudNotifyNodes) {
				session = MessagingService.getInstance().send2Node(
						new InetSocketAddress(node.getIp(),
								ClusterContext.getHeartBeatPort()), message,
						new IoHandlerAdapter());
				session.close(false).awaitUninterruptibly();
			}

			logger.info(
					"finished notifying special Job to end, job master is {}",
					jobController);
		} catch (IOException e) {
			logger.error("", e);
		}
	}
}
